package com.lpssfxy.offline

import com.lpssfxy.offline.entities.{ ProductRecs, Recommendation}
import com.lpssfxy.offline.model.OfflineALSModel
import com.lpssfxy.offline.utils.AppUtils
import org.apache.spark.sql
import org.apache.spark.sql.SparkSession
import org.jblas.DoubleMatrix

object CalculateProductRecListBackup {

  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark = AppUtils.createSparkSession("CalculateProductRecList",AppUtils.getSparkCores)
    // 加载数据
    val ratingRDD = AppUtils.loadRatingData(spark)
    // 计算商品推荐
    val productRecsDF = calculateProductRecommendations(spark, ratingRDD)
    // 将商品推荐结果写入MongoDB
    AppUtils.saveRecommendationsToMongoDB(productRecsDF,AppUtils.MONGODB_PRODUCT_RECS_COLLECTION)
    // 停止SparkSession
    spark.stop()
  }

  /**
   * 计算两个商品之间的余弦相似度
   *
   * @param product1
   * @param product2
   * @return
   */
  private def cosSim(product1: DoubleMatrix, product2: DoubleMatrix): Double = {
    product1.dot(product2) / (product1.norm2() * product2.norm2())
  }

  /**
   * 计算商品推荐列表
   *
   * @param spark     SparkSession对象
   * @param ratingRDD 评分数据的RDD
   * @return          商品推荐列表的DataFrame
   */
  private def calculateProductRecommendations(spark: SparkSession, ratingRDD: org.apache.spark.rdd.RDD[(Int, Int, Double)]): sql.DataFrame = {
    import spark.implicits._
    // 训练 ALS 模型，先任意给定一个参数组合的值
    val (rank, iterations, lambda) = (5, 15, 0.1)
    val model = OfflineALSModel.trainALSModel(ratingRDD, rank, iterations, lambda)
    // 获取商品的特征矩阵，数据格式 RDD[(scala.Int, scala.Array[scala.Double])]
    val productFeatures = model.productFeatures.map { case (productId, features) =>
      (productId, new DoubleMatrix(features))
    }
    // 计算笛卡尔积并过滤合并
    productFeatures.cartesian(productFeatures)
      .filter { case (a, b) => a._1 != b._1 }
      .map { case (a, b) =>
        val simScore = this.cosSim(a._2, b._2) // 求余弦相似度
        (a._1, (b._1, simScore))
      }.filter(_._2._2 > 0.6)
      .groupByKey()
      .map { case (productId, items) =>
        ProductRecs(productId, items.toList.map(x => Recommendation(x._1, x._2)))
      }.toDF()
  }
}

